package com.atguigu.test;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.source.AbstractSource;
import java.io.UnsupportedEncodingException;
import java.util.Random;

/**
 * flume的数据源
 */
public class MySource extends AbstractSource implements Configurable, PollableSource {

    private int i = 0;
    private long time;
    private String prefix;

    /**
     * 核心方法，flume会循环调用这个方法拉取数据，并将数据存放到Channel中
     * @return
     * @throws EventDeliveryException
     */
    public Status process() throws EventDeliveryException {
        Status status = null;
        try {
            //拉取数据
            Event e = getSomeData();
            //向对应的Channel中存放
            ChannelProcessor channelProcessor = getChannelProcessor();
            channelProcessor.processEvent(e);
            status = Status.READY;
        } catch (Exception e) {
            status = Status.BACKOFF;
            e.printStackTrace();
        }
        return status;
    }

    /**
     * 从自定义的源中拉取数据
     * @return
     * @throws InterruptedException
     * @throws UnsupportedEncodingException
     */
    private Event getSomeData() throws InterruptedException, UnsupportedEncodingException {
        i++;
        //生成随机数,最为源数据
        Random random = new Random();
        int data= random.nextInt(10000);
        //获取event
        Event event = new SimpleEvent();
        if (this.i % 5 == 0){
            event.setBody((prefix + (int) data).getBytes("UTF-8"));
        }else {
            event.setBody(String.valueOf(data).getBytes("UTF-8"));
        }
        if (this.i % 10 == 0) {
            //模拟没有拉取到数据
            throw new RuntimeException("逢十报错！");
        }
        Thread.sleep(time);
        return event;
    }

    /**
     * 当没有拉取到数据时的等待时间递增量
     * 第一次没拉取到等待1000ms再次拉取
     * 第二次如果仍然没拉取到就等待2000ms再次拉取
     * 第三次如果仍然没拉取到就等待3000ms再次拉取
     * 如果某次拉取到了数据就从1000ms重新计算
     * @return
     */
    public long getBackOffSleepIncrement() {
        return 1000;
    }

    /**
     * getBackOffSleepIncrement方法的递增最大值
     * @return
     */
    public long getMaxBackOffSleepInterval() {
        return 10000;
    }

    /**
     * 通过配置文件定义我们的Source
     * @param context 配置文件
     */
    public void configure(Context context) {
        time = context.getLong("timeinterval", 100L);
        prefix = context.getString("prefix", "XXXXXX-");
    }
}
